package com.asap.demo.rete;


import com.asap.demo.sourcefunc.MysqlSourceFunction1;
import com.asap.demo.utils.Constants;
import com.asap.demo.utils.Utils;
import com.asap.interf.Action;
import com.asap.rule.StandardEvent;
import com.asap.rule.engine.InferenceEngine;
import com.asap.rule.engine.PatternMatcher;
import com.asap.rule.orm.DbFetcher;
import com.asap.rule.util.PropTransformMap;
import com.asap.rule.util.RuleReader;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.ListTypeInfo;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.*;

public class ReteDemo {

	private static final Logger logger = LoggerFactory.getLogger(ReteDemo.class);

	public static void main(String[] args) throws Exception {

		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);
		env.enableCheckpointing(5000);  //检查点 每5000ms
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

		Properties browseProperties = new Properties();
		browseProperties.put("bootstrap.servers", "192.168.1.25:9093");
		browseProperties.put("group.id", "temporal");
		browseProperties.put("auto.offset.reset", "latest");
		PropTransformMap.getInstance().readConfigMap("/home/asap/wbh/cfg.properties");
		Map<String, String> configMap = new HashMap<String, String>();
		configMap.put(Constants.DB_JDBC_USER, "root");
		configMap.put(Constants.DB_JDBC_PASSWD, "1qazXSW@3edc");
		configMap.put(Constants.DB_JDBC_URL, "jdbc:mysql://192.168.1.234:3306/SSA?useUnicode=true&characterEncoding=utf-8");
		configMap.put(Constants.DB_JDBC_DRIVER, "com.mysql.jdbc.Driver");
		configMap.put(Constants.INITAL_POOL_SIZE, "10");
		configMap.put(Constants.MIN_POOL_SIZE, "5");
		configMap.put(Constants.MAX_IDLE_TIME, "50");
		configMap.put(Constants.MAX_STATE_ELEMENTS, "100");
		configMap.put(Constants.MAX_IDLE_TIME, "60");
		DbFetcher dbFetcher = new DbFetcher(configMap);
		List<String> listRule = RuleReader.readRules(dbFetcher);
		System.out.println("ListRule::" + listRule.size());
		//1、读取mysql的配置消息
		DataStream<List<String>> conf = env.addSource(new MysqlSourceFunction1(dbFetcher));

		//2、创建MapStateDescriptor规则，对广播的数据的数据类型的规则
		MapStateDescriptor<String, List<String>> ruleStateDescriptor = new MapStateDescriptor<>("RulesBroadcastState"
				, BasicTypeInfo.STRING_TYPE_INFO
				, new ListTypeInfo<>(String.class));
		//3、对conf进行broadcast返回BroadcastStream
		final BroadcastStream<List<String>> confBroadcast = conf.broadcast(ruleStateDescriptor);

		DataStream<StandardEvent> kafkaData = env
				.addSource(new FlinkKafkaConsumer<>(
						"flink_pressure_test",
						new SimpleStringSchema(),
						browseProperties
				))
				.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.minutes(1)) {
					@Override
					public long extractTimestamp(String element) {
						return Timestamp.valueOf(Utils.transforDate(StandardEvent.parse(element).getField("CREATE_TIME"))).getTime();
					}
				}).map(new MapFunction<String, StandardEvent>() {
					@Override
					public StandardEvent map(String value) throws Exception {
						StandardEvent standardEvent = StandardEvent.parse(value);
						return standardEvent;
					}
				}).connect(confBroadcast)
				.process(
						new BroadcastProcessFunction<StandardEvent, List<String>, StandardEvent>() {
							private List<String> listRule = new ArrayList<String>();

							MapStateDescriptor<String, List<InferenceEngine>> inferenceEngineDescriptor = new MapStateDescriptor<>("inferenceEngineState"
									, BasicTypeInfo.STRING_TYPE_INFO
									, new ListTypeInfo<>(InferenceEngine.class));
							private boolean flag=false;
							/**
							 * open方法只会执行一次
							 * 可以在这实现初始化的功能
							 * 4、设置keyWords的初始值，否者会报错：java.lang.NullPointerException
							 * @param parameters
							 * @throws Exception
							 */
							@Override
							public void open(Configuration parameters) throws Exception{
								super.open(parameters);
								logger.info("get counter/globalConfig MapState from checkpoint:");
							}
							/**
							 * 6、 处理流中的数据
							 * @param standardEvent
							 * @param readOnlyContext
							 * @param collector
							 * @throws Exception
							 */
							@Override
								public void processElement(StandardEvent standardEvent, ReadOnlyContext readOnlyContext, Collector<StandardEvent> collector) throws Exception {

								List<InferenceEngine> list= readOnlyContext.getBroadcastState(inferenceEngineDescriptor).get("inferenceEngineState");
								PatternMatcher matcher = list.get(0).matcher(standardEvent);
								if (matcher.find()) {
									List<Action> actions = matcher.getActions();
									for (Action action : actions) {
										//standardEvent.setActions(actions);
										collector.collect(standardEvent);
									}
								} else {
									System.out.println("action11111");
								}


							}
							/**
							 * 6、 对广播变量的获取更新
							 * @param strings
							 * @param context
							 * @param collector
							 * @throws Exception
							 */

							@Override
							public void processBroadcastElement(List<String> strings, Context context, Collector<StandardEvent> collector) throws Exception {
								BroadcastState<String, List<InferenceEngine>> broadcastState = context.getBroadcastState(inferenceEngineDescriptor);
								System.out.println("broadcastState2:" + strings.size());
								InferenceEngine engine = null;
								List<InferenceEngine> inferenceEngineList = new ArrayList<>();
								if (strings.get(0).equals("1") || !flag) {
									inferenceEngineList.clear();
									strings.remove(0);
									engine = InferenceEngine.compile(RuleReader.parseRules(strings));
									System.out.println("action update.....:" + strings.size());
									inferenceEngineList.add(engine);
									broadcastState.put("inferenceEngineState", inferenceEngineList);
									flag = true;
								}
							}
						}
				);
		kafkaData.print();
		env.execute("Broadcast test kafka");
	}
}
